package com.hex.ds.hdrs.period.task.service;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.hex.ds.hdrs.agent.service.AgentManager;
import com.hex.ds.hdrs.agent.service.AgentSession;
import com.hex.ds.hdrs.app.service.AppManager;
import com.hex.ds.hdrs.app.service.AppSession;
import com.hex.ds.hdrs.common.constant.Const;
import com.hex.ds.hdrs.common.constant.ErrorCode;
import com.hex.ds.hdrs.common.exception.BizException;
import com.hex.ds.hdrs.common.utils.ErrorUtil;
import com.hex.ds.hdrs.common.utils.HdrsDateUtil;
import com.hex.ds.hdrs.common.utils.ListUtil;
import com.hex.ds.hdrs.cron.po.CronTask;
import com.hex.ds.hdrs.cron.service.CronTaskService;
import com.hex.ds.hdrs.engine.manager.RunningManager;
import com.hex.ds.hdrs.engine.manager.ToDoManager;
import com.hex.ds.hdrs.engine.manager.ToRunManager;
import com.hex.ds.hdrs.engine.service.ProcessorService;
import com.hex.ds.hdrs.period.job.po.JobTask;
import com.hex.ds.hdrs.period.job.po.PeriodJob;
import com.hex.ds.hdrs.period.job.service.PeriodJobService;
import com.hex.ds.hdrs.period.task.converter.IPeriodTaskConfigConverter;
import com.hex.ds.hdrs.period.task.dto.TaskInfo;
import com.hex.ds.hdrs.period.task.po.PeriodTask;
import com.hex.ds.hdrs.period.task.po.PeriodTaskConfig;
import com.hex.ds.hdrs.period.task.util.TaskTrackHelper;
import com.hex.ds.hdrs.period.task.vo.CycleTaskReq;
import com.hex.ds.hdrs.watch.service.WatchService;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;


@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy, @Autowired})
public class TaskFinishService {

    @Autowired
    private RunningManager runningManager;

    @Autowired
    private ToRunManager toRunManager;

    @Autowired
    private ToDoManager toDoManager;


    @Autowired
    private AgentManager agentManager;

    @Autowired
    private AppManager appManager;

    @Autowired
    private ProcessorService processorService;

    @Autowired
    private TaskService taskService;

    @Autowired
    private CronTaskService cronTaskService;

    @Autowired
    private WatchService watchService;

    @Autowired
    private PeriodJobService periodJobService;

    @Autowired
    private PeriodTaskStreamService periodTaskStreamService;

    @Autowired
    private PeriodTaskService periodTaskService;


    @Autowired
    private PeriodTaskConfigService periodTaskConfigService;

    private final IPeriodTaskConfigConverter periodTaskConfigConverter;

    /**
     * @Method: cycleTaskExecBack
     * @Param: [req]
     * @Return: void
     * @Description： 周期任务执行返回
     * @Author： jd.chen
     * @Date： 2021/6/17 18:01
     * @Version： V1.0
     */
    @Retryable(value = {Exception.class },maxAttemptsExpression ="${retry.maxAttempts}",backoff = @Backoff(delayExpression  = "${retry.maxDelay}"))
    public void cycleTaskExecBack(CycleTaskReq req) {
        log.info("任务执行返回【TaskFinishService.cycleTaskExecBack】开始=======》接收到任务编码为: {}", req.toString());
        if (Const.TASK_EXEC_SUCCESS.equals(req.getResStatus())) {
            //执行成功
            this.taskExecSuccDeal(req.getJobCode(), req.getTaskCode(), req.getBizDate(), req);
        } else if (Const.TASK_EXEC_BIZ_FAIL.equals(req.getResStatus())) {
            //业务失败
            this.taskWorkFailDeal(req.getJobCode(), req.getTaskCode(), req.getBizDate(), req);
        } else {
            //执行失败
            this.taskExecFailDeal(req.getJobCode(), req.getTaskCode(), req.getBizDate(), req);
        }
        insertTaskStream(req);
    }

    /**
     * @Method: insertTaskStream <br>
     * @Param: [req] <br>
     * @Return: void <br>
     * @Description：流水表插入数据<br>
     * @Author： gc.zuo<br>
     * @Date： 2021/12/8 9:59 <br>
     * @Version： V1.0<br>
     */
    public void insertTaskStream(CycleTaskReq req) {
        //不管成功还是失败都往流水表插一条数据
        try {
            Map map = new HashMap(16);
            map.put("taskCode", req.getTaskCode());
            map.put("bizDate", req.getBizDate());
            periodTaskStreamService.insert(map);
        } catch (Exception e) {
            log.error("【向流水表插入数据异常: {}", JSONUtil.toJsonStr(req), e);
        }
    }

    /**
     * 定时任务执行返回
     *
     * @param task
     */
    public void cronTaskExecBack(TaskInfo task) {
        log.info("定时任务执行返回【TaskFinishService.cronTaskExecBack】开始=======》");
        try {
            CronTask cronTask = new CronTask();
            cronTask.setPkId(task.getTask().getPkId());
            //cronTask.setExecStTs(task.getExecStTs());
            List<CronTask> list = cronTaskService.qryTaskList(cronTask);
            String execStTs = list.get(0).getExecStTs();
            cronTask.setExecEndTs(task.getTask().getExecEndTs());
            String costTs = HdrsDateUtil.calLastedTime(execStTs, task.getTask().getExecEndTs());
            cronTask.setCostTs(costTs);
            cronTask.setAgentCode(task.getAgentCode());
            cronTask.setUpTs(HdrsDateUtil.getCurTs());
            cronTask.setTaskCode(task.getTaskCode());
            if (Const.TASK_EXEC_SUCCESS.equals(task.getResultTaskStatus())) {
                cronTask.setExecStatus(Const.SUCCESS_CRON);
                // 邮件通知
                watchService.taskFinishNotic(false, task.getAppCode(), task.getTaskCode(),
                        task.getTaskName(), task.getExecStTs(), true);
            } else {
                cronTask.setExecStatus(Const.FAIL_CRON);
                cronTask.setExecFailReason(task.getErrOutput());
                watchService.taskFinishNotic(false, task.getTask().getAppCode(), task.getTaskCode(),
                        task.getTaskName(), task.getExecStTs(), false);
            }
            cronTaskService.updateCronTask(cronTask);
        } catch (Exception e) {
            log.error("定时任务执行返回处理异常【TaskFinishService.cronTaskExecBack】异常: {}", e.getMessage());
        }
    }

    /**
     * 任务完成后，执行失败的任务，需要做的处理
     */
    public void taskExecFailDeal(String jobCode, String taskCode, String bizDate, CycleTaskReq req) {
        String error = StringUtils.isEmpty(req.getErrOutput()) ? req.getResStatus() : req.getErrOutput();
        if (Const.TASK_EXEC_KILL.equals(req.getResStatus())) {
            error = Const.TASK_EXEC_KILL;
        }
        log.info("任务执行处理失败【taskExecFailDeal】jobCode:{},taskCode:{},bizDate:{},error:{}", jobCode, taskCode, bizDate, error);
        try {
            PeriodTask task = runningManager.getTask(taskCode);
            JobTask jobTask = runningManager.getJob(jobCode);
            if (task == null) {
                log.error("【任务处理失败】jobCode:{},taskCode:{},bizDate:{},error:{}", jobCode, taskCode, bizDate, error);
                this.resetTask(jobCode, taskCode, bizDate, req, Const.TASK_FAIL);
            } else {
                //1. 更新任务轨迹
                TaskTrackHelper.addTaskTrack(task, TaskTrackHelper.execFail(String.format(req.getResStatus(), "--", error)));
                //2. 更新任务
                this.updateTaskFail(jobTask, task, error);
                //3. 删除running
                runningManager.removeJob(task, error);
                //4. 更新session 任务
                AppSession appSession = appManager.getValidAppSession(task.getAppCode());
                appSession.delTaskSession(task);
                AgentSession agentSession = agentManager.getAgentSession(task.getAgentCode());
                agentSession.delTaskSession(task);
                //5. 是否需要短信通知
                watchService.taskFinishNotic(true, task.getAppCode(), taskCode, task.getTaskName(), bizDate, false);
            }
        } catch (Exception e) {
            log.error("【taskExecFailDeal】异常:jobCode:{},taskCode:{},bizDate:{},error:{}", jobCode, taskCode, bizDate, e.getMessage());
            this.resetTask(jobCode, taskCode, bizDate, req, Const.TASK_TODO_RUN);
        }
    }

    /**
     * 任务完成后，执行失败的任务，需要做的处理
     */
    public void taskExecFailDeal(JobTask jobTask, PeriodTask task, String error) {
        String jobCode = task.getJobCode();
        String taskCode = task.getTaskCode();
        String bizDate = task.getBizDate();
        log.info("任务返回值处理失败【taskExecFailDeal】jobCode:{},taskCode:{},bizDate:{},error:{}", jobCode, taskCode, bizDate, error);
        try {
            //1. 跟新任务轨迹
            TaskTrackHelper.addTaskTrack(task, TaskTrackHelper.execFail(error));
            //2. 更新任务
            this.updateTaskFail(jobTask, task, error);
            //3. 删除running
            runningManager.removeJob(task, error);
            //4. 更新session 任务
            AppSession appSession = appManager.getValidAppSession(task.getAppCode());
            appSession.delTaskSession(task);
            AgentSession agentSession = agentManager.getAgentSession(task.getAgentCode());
            agentSession.delTaskSession(task);
            //4. 是否需要短信通知
            watchService.taskFinishNotic(true, task.getAppCode(), taskCode, task.getTaskName(), bizDate, false);
        } catch (Exception e) {
            log.error("【taskExecFailDeal】异常: {}", e.getMessage());
            this.resetTask(jobCode, taskCode, bizDate, null, Const.TASK_TODO_RUN);
        }
    }

    /**
     * 任务完成后,成功的任务，需要做的处理
     *
     * @param jobCode
     * @param taskCode
     * @param bizDate
     */
    public void taskExecSuccDeal(String jobCode, String taskCode, String bizDate, CycleTaskReq req) {
        log.info("【任务执行成功处理】开始=======》 jobCode: {} taskCode: {} bizDate: {}", jobCode, taskCode, bizDate);
        JobTask jobTask = runningManager.getJob(jobCode);
        PeriodTask task = runningManager.getTask(taskCode);
        if (jobTask == null || task == null) {
            log.error("【taskExecSuccDeal】【running】不存在任务,作业编码: {} 任务编码: {} 业务日期: {}", jobCode, taskCode, bizDate);
            this.resetTask(jobCode, taskCode, bizDate, req, Const.TASK_SUCCESS);
            return;
        }

        //1. 更新DB
        String nextTaskCode = jobTask.getNextTaskCode(taskCode);
        this.upDb(jobTask, task, nextTaskCode, req);

        //2. 更新内存session的任务
        this.upMemory(task);

        //3. 有效的appSession  触发
        if (appManager.isValidAppSession(task.getAppCode())) {
            this.touchNext(jobTask, task, nextTaskCode);
        } else {
            log.error("【TaskFinishService.taskFinishDeal】appSession失效，未触发");
        }
        //4. 邮件通知
        watchService.taskFinishNotic(true, task.getAppCode(), taskCode, task.getTaskName(), bizDate, true);
    }

    /**
     * 任务业务失败，需要做的处理
     *
     * @param jobCode
     * @param taskCode
     * @param bizDate
     */
    public void taskWorkFailDeal(String jobCode, String taskCode, String bizDate, CycleTaskReq req) {
        log.info("【任务业务失败处理】开始=======》 jobCode: {} taskCode: {} bizDate: {}", jobCode, taskCode, bizDate);
        JobTask jobTask = runningManager.getJob(jobCode);
        PeriodTask task = runningManager.getTask(taskCode);
        try {
            if (jobTask == null || task == null) {
                log.error("【taskWorkFailDeal】【running】不存在任务,作业编码: {} 任务编码: {} 业务日期: {}", jobCode, taskCode, bizDate);
                this.resetTask(jobCode, taskCode, bizDate, req, Const.TASK_TODO_RUN);
                return;
            }
            //1. 更新DB
            this.upWorkFailDb(task, req);
            //2. 更新session 任务
            this.upWorkFailMemory(task);
            //从runningManager中删除
            runningManager.removeJob(task, "任务业务失败");
            toRunManager.addWorkFailJob(jobTask, task, Const.TRACK_TYPE_RUNNING_TO_RUN);

        } catch (Exception e) {
            log.error("【taskCode:{}】【bizDate:{}】【taskWorkFailDeal】异常:{}",
                    taskCode, bizDate, ErrorUtil.errorInfo(e), e);
            throw e;
        } finally {
            log.info("【任务业务失败处理】结束=======》 jobCode: {} taskCode: {} bizDate: {}", jobCode, taskCode, bizDate);
        }
    }

    /**
     * 更新业务处理失败的session及内存
     *
     * @param task
     */
    public void upWorkFailMemory(PeriodTask task) {
        log.info("业务失败更新内存任务【TaskFinishService.upWorkFailMemory】开始=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
        AppSession appSession = appManager.getValidAppSession(task.getAppCode());
        appSession.delTaskSession(task);
        AgentSession agentSession = agentManager.getAgentSession(task.getAgentCode());
        agentSession.delTaskSession(task);
        //1. 更新内存队列
        String curDate = HdrsDateUtil.getCurTs();
        task.setTaskStatus(Const.TASK_TODO_RUN);
        task.setUpTs(curDate);
        log.info("业务失败更新内存任务【TaskFinishService.upWorkFailMemory】结束=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
    }

    public void upMemory(PeriodTask task) {
        log.info("任务执行成功更新内存【TaskFinishService.upMemory】开始=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
        AppSession appSession = appManager.getValidAppSession(task.getAppCode());
        appSession.delTaskSession(task);
        AgentSession agentSession = agentManager.getAliveAgentSession(task.getAgentCode());
        agentSession.delTaskSession(task);
        //1. 更新内存队列
        String curDate = HdrsDateUtil.getCurTs();
        task.setTaskStatus(Const.TASK_SUCCESS);
        task.setUpTs(curDate);
        task.setExecEndTs(curDate);
        //2. 更新内存中任务最新完成时间队列 2024年7月29日15:10:23 最新完成队列已暂时废弃
        //lstFinishManager.updateLstTask(task);
        log.info("任务执行成功更新内存【TaskFinishService.upMemory】结束=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
    }

    /**
     * @Method: upDb
     * @Param: [jobTask, task, nextTaskCode, req]
     * @Return: void
     * @Description：更新DB
     * @Author： jd.chen
     * @Date： 2021/5/28 9:48
     * @Version： V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void upDb(JobTask jobTask, PeriodTask task, String nextTaskCode, CycleTaskReq req) {
        log.info("任务执行成功更新DB【TaskFinishService.upDb】开始=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
        //1. 更新任务状态
        //更新任务轨迹
        TaskTrackHelper.addTaskTrack(task, TaskTrackHelper.execSucc());
        //更新数据库中的任务信息
        this.updateTask(task, req);

        //2. 更新作业状态
        if (StringUtils.isEmpty(nextTaskCode)) {
            //更新数据库和最新完成队列中的任务信息
            this.updateJob(jobTask.getPeriodJob());
            //从内存中的RunningManager中删除已完成的job
            runningManager.removeJob(task, "作业完成");
        }
        log.info("任务执行成功更新DB【TaskFinishService.upDb】结束=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
    }

    /**
     * @Method: upWorkFailDb
     * @Param: [jobTask, task, req]
     * @Return: void
     * @Description： 任务业务失败, 更新DB
     * @Author： jd.chen
     * @Date： 2021/6/17 17:29
     * @Version： V1.0
     */
    public void upWorkFailDb(PeriodTask task, CycleTaskReq req) {
        //1. 更新任务状态
        //   跟新任务轨迹
        TaskTrackHelper.addTaskTrack(task, TaskTrackHelper.execFail("业务失败"));

        log.info("业务失败更新DB任务【TaskFinishService.updateTask】开始=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
        //更新DB
        Map<String, String> map = new HashMap<>(16);
        map.put("taskCode", task.getTaskCode());
        map.put("bizDate", task.getBizDate());
        map.put("taskStatus", Const.TASK_TODO_RUN);
        map.put("trackJson", JSON.toJSONString(task.getTaskTrackMap()));
        if (req != null) {
            map.put("taskExecLog", req.getTaskInfo().getResultTask().getStdOutput());
        }
        taskService.updateTask(map);
        log.info("业务失败更新DB任务【TaskFinishService.updateTask】结束=======》jobCode: {} taskCode: {} bizDate: {}", task.getJobCode(), task.getTaskCode(), task.getBizDate());
    }

    /**
     * 异常  重置任务
     *
     * @param jobCode
     * @param taskCode
     * @param bizDate
     * @param req
     */
    public void resetTask(String jobCode, String taskCode, String bizDate, CycleTaskReq req, String taskStatus) {
        try {
            PeriodTask periodTask = periodTaskService.qryPeriodTaskInfo(taskCode, bizDate);
            if (req != null && !Const.TASK_EXEC_KILL.equals((req.getResStatus()))
                    && periodTask != null && Const.PERIOD_TASK_STATUS_RUN.equals(periodTask.getTaskStatus())) {
                //1. 更新DB状态为1
                this.updateDBTaskFail(jobCode, taskCode, bizDate, req, taskStatus);
            }
            //2. 更新session 任务
            AppSession appSession = appManager.getValidAppSession(req.getAppCode());
            appSession.delTaskSession(taskCode);
            AgentSession agentSession = agentManager.getAgentSession(req.getAgentCode());
            agentSession.delTaskSession(taskCode);

            //3. running中的都删除掉
            runningManager.removeNullTask(jobCode, taskCode);
        } catch (Exception e) {
            log.error("【taskCode:{}】【bizDate:{}】【resetTask】异常:{}",
                    taskCode, bizDate, e.getMessage());
        }
    }

    /**
     * agent任务执行完成，但内存中running队列不存在该作业，任务异常处理
     */
    public void updateDBTaskFail(String jobCode, String taskCode, String bizDate, CycleTaskReq req, String taskStatus) {
        try {
            log.info("【agent任务执行完成，但内存中running队列不存在该作业，任务异常处理】taskCode: {} bizDate: {}", taskCode, bizDate);
            Map<String, String> map = new HashMap<>(16);
            map.put("taskCode", taskCode);
            map.put("bizDate", bizDate);
            map.put("taskStatus", taskStatus);
            //map.put("execEndTs", DateUtil.getCurTs());
            if (toRunManager.isContainsJob(jobCode)) {
                map.put("trackJson", JSON.toJSONString(toRunManager.getJob(jobCode).getTask(taskCode).getTaskTrackMap()));
            }
            if (req != null) {
                map.put("taskExecLog", req.getTaskInfo().getResultTask().getStdOutput());
            }
            taskService.updateTask(map);
        } catch (Exception e) {
            log.error("【taskCode:{}】【bizDate:{}】【updateDBTaskFail】异常: {}",
                    taskCode, bizDate, e.getMessage());
        }
    }

    /**
     * @Method: updateTaskFail
     * @Param: [jobTask, curTask, error]
     * @Return: void
     * @Description： 任务执行失败  更新任务
     * @Author： jd.chen
     * @Date： 2021/6/17 18:01
     * @Version： V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void updateTaskFail(JobTask jobTask, PeriodTask curTask, String error) {
        if (Const.TASK_EXEC_KILL.equals(error)) {
            error = "任务超时";
        }
        String jobCode = curTask.getJobCode();
        String taskCode = curTask.getTaskCode();
        String bizDate = curTask.getBizDate();
        try {
            String curDate = HdrsDateUtil.getCurTs();
            String jobStatus = jobTask.getJobStatus();
            log.info("作业状态{}", jobStatus);

            String editTaskStatus = Const.TASK_FAIL;
            // 更新作业错误信息
            Map<String, String> jobMap = new HashMap<>(16);
            jobMap.put("jobCode", jobCode);
            jobMap.put("bizDate", bizDate);
            jobMap.put("jobStatus", Const.JOB_TO_DO);
            jobMap.put("errorInfo", error);
            periodJobService.updateJob(jobMap);

            log.info("任务执行失败  更新任务【TaskFinishService.updateTaskFail】taskCode: {} bizDate: {}", taskCode, bizDate);
            //1. 更新内存队列
            curTask.setTaskStatus(editTaskStatus);
            curTask.setExecEndTs(curDate);
            curTask.setUpTs(curDate);

            //2. 更新DB
            Map<String, String> taskMap = new HashMap<>(16);
            taskMap.put("taskCode", taskCode);
            taskMap.put("bizDate", bizDate);
            taskMap.put("taskStatus", editTaskStatus);
            taskMap.put("execEndTs", curDate);
            taskMap.put("trackJson", JSON.toJSONString(curTask.getTaskTrackMap()));
            taskMap.put("taskExecLog", error);
            taskMap.put("errCount", "1");
            taskService.updateTask(taskMap);
        } catch (Exception e) {
            log.error("【taskCode:{}】【bizDate:{}】【updateTaskFail】异常: {}",
                    taskCode, bizDate, e.getMessage());
        }
    }

    /**
     * 更新任务
     *
     * @param curTask
     * @return
     */
    public void updateTask(PeriodTask curTask, CycleTaskReq req) {
        try {
            log.info("任务执行成功，更新任务【TaskFinishService.updateTask】开始=======》,taskCode:{},bizDate:{}", curTask.getTaskCode(), curTask.getBizDate());
            //更新DB
            Map<String, String> map = new HashMap<>(16);
            map.put("taskCode", curTask.getTaskCode());
            map.put("bizDate", curTask.getBizDate());
            map.put("taskStatus", Const.TASK_SUCCESS);
            map.put("execEndTs", HdrsDateUtil.getCurTs());
            map.put("trackJson", JSON.toJSONString(curTask.getTaskTrackMap()));
            if (req != null && StrUtil.isNotBlank(req.getTaskInfo().getResultTask().getStdOutput())) {
                map.put("taskExecLog", req.getTaskInfo().getResultTask().getStdOutput());
            } else {
                map.put("taskExecLog", "任务跑批完成");
            }
            taskService.updateTask(map);
            updateTaskConfig(curTask);
            log.info("任务执行成功，更新任务【TaskFinishService.updateTask】结束=======》,taskCode:{},bizDate:{}", curTask.getTaskCode(), curTask.getBizDate());
        } catch (Exception e){
            log.error("任务执行成功，更新任务状态异常，重试=======》,taskCode:{},bizDate:{}", curTask.getTaskCode(), curTask.getBizDate());
            // 此处抛出异常是为了重试机制捕获，勿删
            throw e;
        }
    }

    /**
     * @Method recover <br>
     * @Param e
     * @Param req <br>
     * @Return <br>
     * @Description 重试次数执行完仍异常进入次方法，无调用，勿删!!!<br>
     * @Author jian.qiao<br>
     * @Date 2024/9/25 17:51 <br>
     * @Version V1.0<br>
     */
    @Recover
    public void recover(Exception e,CycleTaskReq req) {
        log.info("重试次数耗尽，任务置为失败,taskCode:{},bizDate:{}", req.getTaskCode(), req.getBizDate());
        JobTask jobTask = runningManager.getJob(req.getJobCode());
        PeriodTask task = runningManager.getTask(req.getTaskCode());
        this.taskExecFailDeal(jobTask, task, "更新任务状态时异常，重试后仍失败：" + e.getMessage());
        // 增加跑批记录
        insertTaskStream(req);
    }

    /**
     * @Method updataTaskConfig
     * @Param periodTask
     * @Return void
     * @Description 任务执行成功后，如果执行时间大于历史时间，则更新历史最大执行时间
     * @Author Yin.Yang
     * @Date 2024/6/12 15:10
     * @Version V1.0
     */
    public void updateTaskConfig(PeriodTask periodTask) {
        try {
            long execStartTime = HdrsDateUtil.strToLong(periodTask.getExecStTs(), HdrsDateUtil.YYYY_MM_DDHHMISS);
            long currentTime = HdrsDateUtil.getMillis(new Date());
            long execTime = (currentTime - execStartTime) / 1000;

            PeriodTaskConfig periodTaskConfig = periodTaskConfigService.queryOne();
            if (periodTaskConfig == null) {
                PeriodTaskConfig initPeriodTaskConfig = periodTaskConfigConverter.toPeriodTaskConfig(Long.toString(execTime),periodTask.getTaskCode(),HdrsDateUtil.getCurTs());
                periodTaskConfigService.add(initPeriodTaskConfig);
                log.info("【updateTaskConfig】初始化initPeriodTaskConfig，设置最大执行时间为: {}", execTime);
                return;
            }

            long maxExecutionTime = Long.parseLong(periodTaskConfig.getMaxExecutionTime());
            if (execTime > maxExecutionTime) {
                periodTaskConfig.setMaxExecutionTime(Long.toString(execTime));
                periodTaskConfig.setTaskCode(periodTask.getTaskCode());
                periodTaskConfig.setUpTs(HdrsDateUtil.getCurTs());
                periodTaskConfigService.update(periodTaskConfig);
                log.info("【updateTaskConfig】更新PeriodTaskConfig，新的最大执行时间为: {}", execTime);
            }
        } catch (Exception e) {
            log.error("【updateTaskConfig】更新任务配置时发生异常: {}", e.getMessage(), e);
        }
    }


    /**
     * 更新作业
     *
     * @param job
     * @return
     */
    public void updateJob(PeriodJob job) {
        log.info("任务执行成功，更新作业【TaskFinishService.updateJob】开始=======》,jobCode:{},bizDate:{}", job.getJobCode(), job.getBizDate());
        //1. 设置作业信息
        String curDate = HdrsDateUtil.getCurTs();
        job.setJobStatus(Const.JOB_DO_SUCCESS);
        job.setUpTs(curDate);
        job.setJobEndTs(curDate);
        //2. 更新DB
        Map<String, String> map = new HashMap<>(16);
        map.put("jobCode", job.getJobCode());
        map.put("bizDate", job.getBizDate());
        map.put("jobEndTs", curDate);
        map.put("jobStatus", Const.JOB_DO_SUCCESS);
        map.put("upTs", curDate);
        taskService.updateJob(map);
        //3. 更新内存中作业最新完成时间队列 2024年8月5日11:03:38 暂时废除最新完成队列，不在更新
        //lstFinishManager.updateLstJob(job);
        log.info("任务执行成功，更新作业【TaskFinishService.updateJob】结束=======》,jobCode:{},bizDate:{}", job.getJobCode(), job.getBizDate());
    }

    /**
     * 触发
     *
     * @param jobTask
     */
    public void touchNext(JobTask jobTask, PeriodTask curTask, String nextTaskCode) {
        log.info("触发【TaskFinishService.touchNext】开始=======》");
        try {
            if (jobTask.isFinish()) {
                // 触发下个日期的本作业
                this.touchNextDateJob(jobTask);
            } else {
                //1. 触发本作业下的任务
                this.touchNextTaskInJob(jobTask, curTask, nextTaskCode);
            }
            //2. 触发其他作业的下游任务
//            this.touchNextTaskByTaskCode(jobTask, curTask);
        } catch (Exception e) {
            log.error("【nextTaskCode:{}】【touchNext】异常: {}", nextTaskCode, e.getMessage());
        }
    }

    /**
     * 触发任务的下游任务
     *
     * @param jobTask
     */
    public void touchNextTaskByTaskCode(JobTask jobTask, PeriodTask curTask) {
        log.info("触发任务的下游任务【TaskFinishService.touchNextTaskByTaskCode】curTask:【{}】【{}】", curTask.getTaskCode(), curTask.getBizDate());
        try {
            List<PeriodTask> downTaskList = periodTaskService.queryListByDependTask(curTask.getTaskCode(), curTask.getBizDate());
            if (ListUtil.isEmpty(downTaskList)) {
                log.debug("【TaskFinishService.touchNextTaskByTaskCode】无下游触发任务======");
                return;
            }
            for (PeriodTask downTask : downTaskList) {
                if (downTask.getJobCode().equals(jobTask.getJobCode())) {
                    //1. 本作业下的下游任务
                    continue;
                }
                touchNextTaskOtherJob(downTask.getJobCode(), downTask.getTaskCode());
            }
        } catch (BizException e) {
            log.error("【touchNextTaskByTaskCode】异常:{} ", e.getErrMsg(), e);
        }
    }

    /**
     * 本作业下的下游任务
     *
     * @param jobTask
     * @param curTask
     */
    public void touchNextTaskInJob(JobTask jobTask, PeriodTask curTask, String nextTaskCode) {
        log.info("触发任务的下游任务【TaskFinishService.touchNextTaskInJob】开始======》");
        // 获取下一个任务
        PeriodTask downTask = jobTask.getTask(nextTaskCode);
        try {
            if (downTask == null) {
                log.warn("下一任务【{}】为null", nextTaskCode);
                List<PeriodTask> sortList = jobTask.getPeriodTasks();
                for (PeriodTask t : sortList) {
                    if (t == null) {
                        log.warn("t is null");
                    } else {
                        log.warn("code:{}===============bizDate:{}", t.getTaskCode(), t.getBizDate());
                    }
                }
                downTask = periodTaskService.qryPeriodTaskInfo(nextTaskCode, curTask.getBizDate());
                if (downTask == null) {
                    runningManager.removeJob(curTask, "依赖触发异常");
                    throw new BizException(ErrorCode.ERROR_823001);
                }
            }
            // 判断状态
            String downTaskStatus = downTask.getTaskStatus();
            if (Const.TASK_TODO_RUN.equals(downTaskStatus) || Const.TASK_RUNNING.equals(downTaskStatus)) {
                // 检查互斥
                this.processorService.checkMutex(downTask);
                // 检查上游依赖
                this.processorService.checkUpDepend(downTask);
                // 上游依赖已经完成，直接添加到可执行队列
                log.info("【TaskFinishService.touchNextTaskInJob】上游依赖已经完成，正在运行添加到可执行======任务号:{}", downTask.getTaskCode());
                if (Const.ALL_LOADER_MARK.equals(toDoManager.getMark())) {
                    toDoManager.addJobToDeltaMap(jobTask);
                }
                runningManager.touchNextTaskToToRun(jobTask, downTask, curTask, toRunManager);
            } else {
                throw new BizException(ErrorCode.ERROR_823000);
            }
        } catch (BizException e) {
            log.error("【nextTaskCode:{}】【touchNextTaskInJob】异常:{}", nextTaskCode, e.getErrMsg(), e);
            runningManager.runningToToDo(jobTask, curTask, toDoManager);
        } catch (Exception e) {
            log.error("【nextTaskCode:{}】【touchNextTaskInJob】未知异常:{}", nextTaskCode, e.getMessage() ,e);
            runningManager.runningToToDo(jobTask, curTask, toDoManager);
        }
    }

    /**
     * 其他作业下的下游任务
     *
     * @param jobCode
     * @param downTaskCode
     */
    public void touchNextTaskOtherJob(String jobCode, String downTaskCode) {
        log.info("其他作业下的下游任务【TaskFinishService.touchNextTaskOtherJob】开始======》");
        try {
            if (!toDoManager.isContainsJob(jobCode)) {
                log.debug("【TaskFinishService.touchNextTaskOtherJob】任务不在待处理======任务号: {}", downTaskCode);
                return;
            }

            JobTask jobTask = toDoManager.getJob(jobCode);
            PeriodTask curTask = jobTask.getCurTask();
            if (curTask == null) {
                log.error("当前任务为空【touchNextTaskOtherJob】【jobTask.getCurTask】job: {} bizDate: {}", jobTask.getJobCode(), jobTask.getBizDate());
                return;
            } else {
                if (!downTaskCode.equals(curTask.getTaskCode())) {
                    //下游依赖的任务在其作业中不是当前任务
                    log.debug("【TaskFinishService.touchNextTaskOtherJob】该job的当前任务不是这个触发任务======任务号: {}", downTaskCode);
                    return;
                }

                this.processorService.checkUpDepend(curTask);
                //上游依赖已经完成，直接添加到可执行队列
                log.info("【TaskFinishService.touchNextTaskOtherJob】上游依赖已经完成，待处理添加到可执行======任务号: {}", downTaskCode);
                toDoManager.toDoToToRun(jobTask, curTask, Const.TRACK_TYPE_TOUCH_TO_RUN, toRunManager);
            }
        } catch (BizException e) {
            log.error("【downTaskCode:{}】【touchNextTaskOtherJob】异常:{} ", downTaskCode, e.getErrMsg(), e);
        }
    }

    /**
     * 获取下一日期的该作业实例
     *
     * @param jobTask
     */
    public void touchNextDateJob(JobTask jobTask) {
        String jobCode = jobTask.getJobCode();
        String bizDate = jobTask.getBizDate();
        log.info("获取下一日期的该作业实例【TaskFinishService.touchNextDateJob】jobCode: {} bizDate: {}", jobCode, bizDate);
        try {
            //1. 判断内存中是否存在
            if (checkJobInMemory(jobCode)) {
                log.warn("内存中已存在下一日期作业，不触发【touchNextDateJob】jobCode: {} bizDate: {}", jobCode, bizDate);
                return;
            }

            //2. 内存中没有，从db中获取jobCode业务日期最早未完成的作业
            PeriodJob job = taskService.qryNextDateJob(jobCode);
            if (null != job) {
                String jobBizDate = job.getBizDate();
                // 判断最早未完成作业的业务日期是否在当天之前，是则可触发
                if (!cn.hutool.core.date.DateUtil.parse(jobBizDate)
                        .after(cn.hutool.core.date.DateUtil.parse(HdrsDateUtil.getCurrentDate()))) {
                    List<PeriodTask> taskList = taskService.qryNextDateTask(jobCode, jobBizDate);
                    if (!ListUtil.isEmpty(taskList)) {
                        String taskBizDate = taskList.get(0).getBizDate();
                        if (jobBizDate.equals(taskBizDate)) {
                            JobTask nextDateJobTask = new JobTask(job, taskList);
                            // 3. 添加到待处理队列
                            log.info("【touchNextDateJob】下一个作业【{}】【{}】", nextDateJobTask.getJobCode(), nextDateJobTask.getBizDate());
                            toDoManager.addJob(nextDateJobTask, Const.TRACK_TYPE_TOUCH_TO_DO);
                        } else {
                            log.warn("业务日期异常【jobTask】{} jobBizDate: {} taskBizDate: {}", jobCode, jobBizDate, taskBizDate);
                        }
                    } else {
                        log.warn("下一个日期的任务实例为null【touchNextDateJob】job:【{}】【{}】", jobCode, jobBizDate);
                    }
                }
            }
        } catch (BizException e) {
            log.error("【jobCode:{}】【bizDate:{}】【touchNextDateJob】异常：{}", jobCode, bizDate, e.getErrMsg(), e);
        }
    }

    public boolean checkJobInMemory(String jobCode) {
        JobTask jobTask = null;
        if (toDoManager.isContainsJob(jobCode)) {
            jobTask = toDoManager.getJob(jobCode);
        } else if (toRunManager.isContainsJob(jobCode)) {
            jobTask = toRunManager.getJob(jobCode);
        } else if (runningManager.isContainsJob(jobCode)) {
            jobTask = runningManager.getJob(jobCode);
        }

        if (jobTask != null) {
            return true;
        } else {
            return false;
        }
    }
}
